/*
Copyright 2023 The Nuclio Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package trigger

import (
	"strconv"
	"strings"
	"sync/atomic"
	"time"

	"github.com/nuclio/nuclio/pkg/functionconfig"
	"github.com/nuclio/nuclio/pkg/processor/runtime"
	"github.com/nuclio/nuclio/pkg/processor/statistics"
	"github.com/nuclio/nuclio/pkg/processor/util/partitionworker"

	"github.com/nuclio/errors"
	"github.com/nuclio/logger"
)

type DurationConfigField struct {
	Name    string
	Value   string
	Field   *time.Duration
	Default time.Duration
}

type AnnotationConfigField struct {
	Key             string
	ValueString     *string
	ValueListString []string
	ValueInt        *int
	ValueUInt64     *uint64
	ValueBool       *bool
}

type Configuration struct {
	*functionconfig.Trigger

	// the runtime configuration, for reference
	RuntimeConfiguration *runtime.Configuration

	// a unique trigger ID
	ID string
}

func NewConfiguration(id string,
	triggerConfiguration *functionconfig.Trigger,
	runtimeConfiguration *runtime.Configuration) (*Configuration, error) {

	configuration := &Configuration{
		Trigger:              triggerConfiguration,
		RuntimeConfiguration: runtimeConfiguration,
		ID:                   id,
	}

	// set defaults
	if configuration.NumWorkers == 0 {
		configuration.NumWorkers = 1
	}

	if triggerConfiguration.WorkerTerminationTimeout == "" {
		triggerConfiguration.WorkerTerminationTimeout = functionconfig.DefaultWorkerTerminationTimeout
	}

	workerTerminationTimeout, err := time.ParseDuration(triggerConfiguration.WorkerTerminationTimeout)
	if err != nil {
		return nil, errors.New("Failed to parse worker termination timeout from trigger configuration")
	}
	runtimeConfiguration.WorkerTerminationTimeout = workerTerminationTimeout

	return configuration, nil
}

// PopulateConfigurationFromAnnotations allows setting configuration via annotations, for experimental settings
func (c *Configuration) PopulateConfigurationFromAnnotations(annotationConfigFields []AnnotationConfigField) error {
	var err error

	for _, annotationConfigField := range annotationConfigFields {
		annotationValue, annotationKeyExists := c.RuntimeConfiguration.Meta.Annotations[annotationConfigField.Key]
		if !annotationKeyExists {
			continue
		}

		switch {
		case annotationConfigField.ValueString != nil:
			*annotationConfigField.ValueString = annotationValue
		case annotationConfigField.ValueInt != nil:
			*annotationConfigField.ValueInt, err = strconv.Atoi(annotationValue)
			if err != nil {
				return errors.Wrapf(err, "Annotation %s must be numeric", annotationConfigField.Key)
			}
		case annotationConfigField.ValueBool != nil:
			*annotationConfigField.ValueBool, err = strconv.ParseBool(annotationValue)
			if err != nil {
				return errors.Wrapf(err, "Annotation %s must represent boolean", annotationConfigField.Key)
			}
		case annotationConfigField.ValueListString != nil:
			annotationConfigField.ValueListString = strings.Split(annotationValue, ",")
		case annotationConfigField.ValueUInt64 != nil:
			*annotationConfigField.ValueUInt64, err = strconv.ParseUint(annotationValue, 10, 64)
			if err != nil {
				return errors.Wrapf(err, "Annotation %s must be positive numeric (uint64)", annotationConfigField.Key)
			}
		}
	}

	return nil
}

// ParseDurationOrDefault parses a duration string into a time.duration field. if empty, sets the field to the default
func (c *Configuration) ParseDurationOrDefault(durationConfigField *DurationConfigField) error {
	if durationConfigField.Value == "" {
		*durationConfigField.Field = durationConfigField.Default
		return nil
	}

	parsedDurationValue, err := time.ParseDuration(durationConfigField.Value)
	if err != nil {
		return errors.Wrapf(err, "Failed to parse %s", durationConfigField.Name)
	}

	*durationConfigField.Field = parsedDurationValue

	return nil
}

func (c *Configuration) PopulateExplicitAckMode(logger logger.Logger, explicitAckModeValue string,
	triggerConfigurationExplicitAckMode functionconfig.ExplicitAckMode) error {
	switch explicitAckModeValue {
	case string(functionconfig.ExplicitAckModeEnable):
		c.ExplicitAckMode = functionconfig.ExplicitAckModeEnable
	case string(functionconfig.ExplicitAckModeExplicitOnly):
		c.ExplicitAckMode = functionconfig.ExplicitAckModeExplicitOnly
	default:

		// default explicit ack mode to 'disable' if not set
		if triggerConfigurationExplicitAckMode != "" {
			c.ExplicitAckMode = triggerConfigurationExplicitAckMode
		} else {
			c.ExplicitAckMode = functionconfig.ExplicitAckModeDisable
		}
	}

	if c.ExplicitAckMode != functionconfig.ExplicitAckModeDisable {
		if !functionconfig.RuntimeSupportExplicitAck(c.RuntimeConfiguration.Spec.Runtime) {
			logger.WarnWith("Explicit Ack is not supported for the configured runtime. "+
				"Setting explicitAck mode to `disable`",
				"runtime", c.RuntimeConfiguration.Spec.Runtime)
			c.ExplicitAckMode = functionconfig.ExplicitAckModeDisable
		}
	}
	return nil
}

func (c *Configuration) ResolveWorkerAllocationMode(modeFromAttributes, modeFromAnnotation partitionworker.AllocationMode) partitionworker.AllocationMode {

	// prioritize attribute over annotation
	if modeFromAttributes != "" {
		return modeFromAttributes
	}
	if modeFromAnnotation != "" {
		return modeFromAnnotation
	}

	// default to pool
	return partitionworker.AllocationModePool
}

type Statistics struct {
	EventsHandledSuccessTotal atomic.Uint64
	EventsHandledFailureTotal atomic.Uint64
	WorkerAllocatorStatistics statistics.AllocatorStatistics
}

// UnsafeStatistics is the same object as Statistics but thread unsafe
// we need it to create a copy of Statistics (which contains atomic vars, which are not possible to copy)
// and since we don't need statistics to be thread-safe during metrics gathering, we can avoid any synchronisation costs
type UnsafeStatistics struct {
	EventsHandledSuccessTotal uint64
	EventsHandledFailureTotal uint64
	WorkerAllocatorStatistics statistics.AllocatorStatistics
}

func (s *UnsafeStatistics) DiffFrom(prev *UnsafeStatistics) *UnsafeStatistics {
	workerAllocatorStatisticsDiff := s.WorkerAllocatorStatistics.DiffFrom(&prev.WorkerAllocatorStatistics)
	currEventsHandledSuccessTotal := s.EventsHandledSuccessTotal
	currEventsHandledFailureTotal := s.EventsHandledFailureTotal
	prevEventsHandledSuccessTotal := prev.EventsHandledSuccessTotal
	prevEventsHandledFailureTotal := prev.EventsHandledFailureTotal

	diffStatistics := &UnsafeStatistics{
		WorkerAllocatorStatistics: workerAllocatorStatisticsDiff,
		EventsHandledSuccessTotal: currEventsHandledSuccessTotal - prevEventsHandledSuccessTotal,
		EventsHandledFailureTotal: currEventsHandledFailureTotal - prevEventsHandledFailureTotal,
	}
	return diffStatistics
}

type Secret struct {
	Contents string
}
